package com.ku.rocketmq.quickstart.transaction;

import com.ku.rocketmq.constants.Const;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.Charset;
import java.util.concurrent.*;

/**
 * @program: rocketmqDemo
 * @description: 事务消息生产者
 * @author: JokerKu
 * @create: 2020-05-06 14:17
 **/
public class TransactionProvider {

    public static void main(String[] args) throws MQClientException {
        //1.创建事务消息生产者
        TransactionMQProducer producer = new TransactionMQProducer("provider_tx_group");
        //创建一个线程池来处理业务
        ExecutorService executorService = new ThreadPoolExecutor(
                2,
                5,
                100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("text_tx_provider" + "-check-thread");
                        return thread;
                    }
                }
        );
        //2.设置nameServer
        producer.setNamesrvAddr(Const.NAMESERVER_ADDR);
        //3.设置线程池
        producer.setExecutorService(executorService);

        //4.设置事务监听器
        //这个方法中调用的匿名内部类的方法要做两个事情：
        producer.setTransactionListener(new TransactionListener() {
            //1.执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                System.out.println("执行本地业务...");
                //回调参数
                String cllArg = (String) o;
                System.out.println("callArg" + cllArg);
                //tx.begin
                //业务逻辑

                //tx.commit
                return LocalTransactionState.COMMIT_MESSAGE;//提交消息
                //return LocalTransactionState.ROLLBACK_MESSAGE;//回滚消息--删除掉MQ中的消息
                //return LocalTransactionState.UNKNOW;//重试
            }

            //2.做check回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("回调检查.....");
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        //5.启动producer
        producer.start();
        //6.封装消息
        Message message = new Message("test_tx_topic", "TagA", "KeyA", "Hello World".getBytes(Charset.defaultCharset()));
        //7.发送消息
        producer.sendMessageInTransaction(message, "我是回调的参数");

//        producer.shutdown();
    }
}
